package cn.net.yugu.doraemon.kafka.configuration;

import cn.net.yugu.doraemon.kafka.config.YgKafkaProperties;
import cn.net.yugu.doraemon.kafka.config.YgKafkaSaslConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * @author wanghongli
 * @date 2023/4/1 15:47
 * @description KafkaConsumerAutoConfiguration
 **/
@Configuration
@EnableConfigurationProperties(YgKafkaProperties.class)
public class YgKafkaAutoConfiguration {
    @Bean
    @ConditionalOnProperty(prefix = "yg.kafka", value = "producer-enable", havingValue = "true")
    public KafkaTemplate<String, String> kafkaTemplate(YgKafkaProperties properties) {
        return new KafkaTemplate<>(producerFactory(properties));
    }

    public ProducerFactory<String, String> producerFactory(YgKafkaProperties properties) {
        return new DefaultKafkaProducerFactory<>(producerConfigs(properties));
    }

    public Map<String, Object> producerConfigs(YgKafkaProperties propertiesConfig) {
        Map<String, Object> props = new HashMap<>(15);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, propertiesConfig.getBootstrapServer());
        // Kafka 消息的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, propertiesConfig.getProducerKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, propertiesConfig.getProducerValueSerializer());

        //hostname校验改成空
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, propertiesConfig.getSslEndpointIdentificationAlgorithm());

        props.put(ProducerConfig.RETRIES_CONFIG, propertiesConfig.getProducerRetries());
        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, propertiesConfig.getProducerReconnectBackoffMs());
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, propertiesConfig.getProducerMaxBlockMs());
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
        if (StringUtils.hasLength(propertiesConfig.getSecurityProtocol())) {
            // ssl 认证
            YgKafkaSaslConfig.kafkaSaslConfig(props, propertiesConfig);
        }
        return props;
    }

    /**
     * 消费配置
     */
    public Map<String, Object> consumerConfigs(YgKafkaProperties propertiesConfig) {
        Map<String, Object> props = new HashMap<>(15);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, propertiesConfig.getBootstrapServer());
        // Kafka 消息的序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, propertiesConfig.getConsumerKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, propertiesConfig.getConsumerValueDeserializer());
        if (StringUtils.hasLength(propertiesConfig.getConsumerGroupId())) {
            props.put(ConsumerConfig.GROUP_ID_CONFIG, propertiesConfig.getConsumerGroupId());
        }
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, propertiesConfig.getConsumerAutoCommit());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, propertiesConfig.getConsumerAutoCommitInterval());
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, propertiesConfig.getConsumerSessionTimeout());
        //hostname校验改成空
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, propertiesConfig.getSslEndpointIdentificationAlgorithm());
        if (StringUtils.hasLength(propertiesConfig.getSecurityProtocol())) {
            // ssl 认证
            YgKafkaSaslConfig.kafkaSaslConfig(props, propertiesConfig);
        }
        return props;
    }

    /**
     * 消费消费工厂
     */
    public ConsumerFactory<Integer, String> consumerFactory(YgKafkaProperties properties) {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(properties));
    }

    @Bean
    @ConditionalOnProperty(prefix = "yg.kafka", value = "consumer-enable", havingValue = "true")
    public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory(YgKafkaProperties propertiesConfig) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory(propertiesConfig));
        // 多线程消费
        factory.setConcurrency(propertiesConfig.getConsumerConcurrency());
        return factory;
    }
}
